package com.eninus.notification.ios;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;

import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import javax.net.ssl.SSLEngineResult;
import javax.net.ssl.SSLEngineResult.HandshakeStatus;
import javax.net.ssl.SSLException;
import javax.net.ssl.SSLSession;
import javax.net.ssl.TrustManagerFactory;

public class PushClientThread extends Thread {
    int instanceID;
    
	private ByteBuffer handshakeByteBuffer;

	ByteBuffer outNetByteBuffer;

	ByteBuffer inAppByteBuffer;
	ByteBuffer inNetByteBuffer;

	SSLContext iSSLContext;
	SSLEngine iSSLEngine;
	SSLSession iSSLSession;
	HandshakeStatus handShakeStatus;

	String serverName;
	int port;

	Selector clientSelector;
	SocketChannel socketChannel;
	ChannelInfo clientChannelInfo;
	SelectionKey clientSelectionKey;
	
	boolean isHandshaked;
	
	boolean isCloseFlag;
	boolean isSelectNegative1;
	boolean isResponseClosed;
	boolean isReadNegative1;
	boolean isWriteNegative1;
	
	
	boolean isSelectException;
	boolean isConnectException;
	boolean isBeginHandshakeException;
	boolean isReadException;
	boolean isWriteException;
	
	boolean isWrapException;
	boolean isUnwrapException;
	

	ConcurrentLinkedQueue<IOSNotificationPacket> notificationCLinkedQueue;
	ConcurrentLinkedQueue<IOSNotificationPacket> sendTimeCLinkedQueue;
	ConcurrentLinkedQueue<IOSErrorResponsePacket> errorResponseCLinkedQueue;

	public PushClientThread(int pInstanceID,
			KeyManagerFactory pKeyManagerFactory,
			TrustManagerFactory pTrustManagerFactory,
			String pServerName,
			int pPort,
			ConcurrentLinkedQueue<IOSNotificationPacket> pNotificationCLinkedQueue,
			ConcurrentLinkedQueue<IOSNotificationPacket> pSendTimeCLinkedQueue,ConcurrentLinkedQueue<IOSErrorResponsePacket> pErrorResponseCLinkedQueue)
			throws Exception {
		instanceID=pInstanceID;
		
		serverName=pServerName;
		port=pPort;
		
		notificationCLinkedQueue = pNotificationCLinkedQueue;
		sendTimeCLinkedQueue = pSendTimeCLinkedQueue;
		errorResponseCLinkedQueue=pErrorResponseCLinkedQueue;

		iSSLContext = SSLContext.getInstance("TLS");
		iSSLContext.init(pKeyManagerFactory.getKeyManagers(),
				pTrustManagerFactory.getTrustManagers(), null);

		iSSLEngine = iSSLContext.createSSLEngine(serverName, port);
		iSSLEngine.setUseClientMode(true);

		clientSelector = Selector.open();

		socketChannel = SocketChannel.open();
		// socketChannel.socket().setSendBufferSize(arg0)
		// socketChannel.socket().setReceiveBufferSize(size)
		socketChannel.configureBlocking(false);
		socketChannel.connect(new InetSocketAddress(serverName, port));

		clientChannelInfo = new ChannelInfo();
		clientChannelInfo.connetingTimeMillis = System.currentTimeMillis();

		clientSelectionKey = socketChannel.register(clientSelector,
				socketChannel.validOps(), clientChannelInfo);

		iSSLSession = iSSLEngine.getSession();

		handshakeByteBuffer = ByteBuffer.allocateDirect(0);

		outNetByteBuffer = ByteBuffer.allocateDirect(iSSLSession
				.getPacketBufferSize());
		inAppByteBuffer = ByteBuffer.allocate(iSSLSession
				.getApplicationBufferSize());
		inNetByteBuffer = ByteBuffer.allocateDirect(iSSLSession
				.getPacketBufferSize());
	}

	public void run() {

		while (true) {
			try {
				int selectedNum = clientSelector.selectNow();
				if(-1==selectedNum){
					isSelectNegative1=true;
					System.out.println("****** Selector error.");
				}
			} catch (IOException e) {
				isSelectException=true;
				e.printStackTrace();
			}

			Set<SelectionKey> selectedKeys = clientSelector.selectedKeys();
			Iterator<SelectionKey> keysIterator = selectedKeys.iterator();
			while (keysIterator.hasNext()) {

				SelectionKey key = (SelectionKey) keysIterator.next();
				SocketChannel keySocketChannel = (SocketChannel) key.channel();

				handShakeStatus = iSSLEngine.getHandshakeStatus();
				//System.out.println(handShakeStatus);
				
				if (key.isConnectable()) {
					try {
						if (keySocketChannel.finishConnect()) {
							System.out.println("****** Connected,---instanceID:"+instanceID+","+new Date());
							System.out.println("****** ServerIP:"+keySocketChannel.socket().getInetAddress()+ ":"+ keySocketChannel.socket().getPort() +" ,localPort:"+keySocketChannel.socket().getLocalPort()+",---instanceID:"+instanceID+","+new Date());
							iSSLEngine.beginHandshake();
						}
					} catch (SSLException e) {
						isBeginHandshakeException=true;
						e.printStackTrace();
					} catch (IOException e) {
						isConnectException=true;
						e.printStackTrace();
					}
				}

				if (key.isReadable()) {
					if (inNetByteBuffer.position() == inNetByteBuffer.limit()) {
						inNetByteBuffer.clear();
					}
					int readInNetDataSize = 0;
					
						try {
							readInNetDataSize = keySocketChannel
									.read(inNetByteBuffer);
						} catch (IOException e) {
							isReadException=true;
							e.printStackTrace();
						}
					
					System.out.println("===>>> Read "+readInNetDataSize+ " bytes.");
					if (-1 == readInNetDataSize) {
						System.out.println("XXXXXX Read error.");
						isReadNegative1 = true;
						// -iSSLEngine.closeInbound();
					}
				}
				//
				if (handShakeStatus == SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING) {

					inNetByteBuffer.flip();
					if (inNetByteBuffer.hasRemaining()) {
						try {
							SSLEngineResult resUnwrap = iSSLEngine.unwrap(
									inNetByteBuffer, inAppByteBuffer);

							switch (resUnwrap.getStatus()) {
							case BUFFER_OVERFLOW:
								System.out
										.println("XXXXXX Unwrap buffer_overflow.");
								break;
							case CLOSED:
								System.out
										.println("XXXXXX The server closed the socket.");
								break;
							case BUFFER_UNDERFLOW:
								// System.out.println("Unwrap BUFFER_UNDERFLOW");
								break;
							case OK:
							}
						} catch (SSLException e) {
							isUnwrapException=true;
							e.printStackTrace();
						}
					}
					inNetByteBuffer.compact();

					//
					inAppByteBuffer.flip();
					if (inAppByteBuffer.limit()>=5) {
							IOSErrorResponsePacket iosErrorResponsePacket=new IOSErrorResponsePacket();
							iosErrorResponsePacket.command=inAppByteBuffer.get();
							iosErrorResponsePacket.status=inAppByteBuffer.get();
							iosErrorResponsePacket.identifier=inAppByteBuffer.getInt();
							errorResponseCLinkedQueue.add(iosErrorResponsePacket);
							//apple's server closed the connection.
							if(10==iosErrorResponsePacket.status){
								isResponseClosed=true;
							}
					}
					inAppByteBuffer.compact();

					if (key.isWritable()) {// &&key.isValid()

						if (outNetByteBuffer.position() == outNetByteBuffer
								.limit()) {
							outNetByteBuffer.clear();
						}
						IOSNotificationPacket iosNotificationPacket = notificationCLinkedQueue
								.poll();
						if (iosNotificationPacket != null) {
							iosNotificationPacket.bbuffer.flip();
							try {
								//SSLEngineResult resWrap = 
								iSSLEngine.wrap(
										iosNotificationPacket.bbuffer,
										outNetByteBuffer);
							} catch (SSLException e) {
								isWrapException=true;
								e.printStackTrace();
							}
						}

						outNetByteBuffer.flip();
						int writeOutNetDataSize = 0;
						while (outNetByteBuffer.hasRemaining()) {
							
							try {
								writeOutNetDataSize = keySocketChannel
										.write(outNetByteBuffer);
								System.out.println("<<<=== Wrote "+writeOutNetDataSize+ " bytes for notification,---instanceID:"+instanceID+","+new Date());
							} catch (IOException e) {
								isWriteException=true;
								e.printStackTrace();
								break;
							}
							if (-1==writeOutNetDataSize) {
								// handle closed channel
								isWriteNegative1 = true;
								System.out.println("XXXXXX Write error.");
								break;
							}
						}
						if (iosNotificationPacket != null&&writeOutNetDataSize!=-1){
							//ChannelInfo keyChannelInfo=  (ChannelInfo)key.attachment();
							//following for client
							long currentTimeMillis=System.currentTimeMillis();
							
							iosNotificationPacket.millisecondsSince1970=currentTimeMillis;
							
							sendTimeCLinkedQueue.add(iosNotificationPacket);
							
							clientChannelInfo.sentCount=clientChannelInfo.sentCount+1;
							clientChannelInfo.lastSendTimeMillis=currentTimeMillis;
						}
					}
				} else {
					switch (handShakeStatus) {
					case NEED_WRAP:
						if (outNetByteBuffer.position() == outNetByteBuffer
								.limit()) {
							outNetByteBuffer.clear();
						}
						try {
							//SSLEngineResult resWrap = 
									iSSLEngine.wrap(
									handshakeByteBuffer, outNetByteBuffer);
						} catch (SSLException e) {
							isWrapException=true;
							e.printStackTrace();
						}

						outNetByteBuffer.flip();
						while (outNetByteBuffer.hasRemaining()) {
							int writeOutNetDataSize = 0;
							try {
								writeOutNetDataSize = keySocketChannel
										.write(outNetByteBuffer);
								System.out.println("<<<=== Wrote "+writeOutNetDataSize+ " bytes." );
							} catch (IOException e) {
								isWrapException=true;
								e.printStackTrace();
								break;
							}
							if (-1==writeOutNetDataSize) {
								// handle closed channel
								isWriteNegative1 = true;
								System.out.println("XXXXXX Write error.");
								break;
							}
						}
						outNetByteBuffer.compact();

						break;
					case NEED_UNWRAP:
						inNetByteBuffer.flip();
						// if(inNetByteBuffer.hasRemaining()){
						try {
							SSLEngineResult resUnwrap = iSSLEngine.unwrap(
									inNetByteBuffer, inAppByteBuffer);
							if(resUnwrap.getHandshakeStatus()==HandshakeStatus.FINISHED){
									isHandshaked=true;
									System.out.println("****** Handshake finished,---instanceID:"+instanceID+","+new Date());
							}
							switch (resUnwrap.getStatus()) {
							case BUFFER_OVERFLOW:
								System.out
										.println("XXXXXX Unwrap buffer_overflow.");
								break;
							case CLOSED:
								System.out
										.println("XXXXXX The server closed the socket.");
								break;
							case BUFFER_UNDERFLOW:
								// System.out.println("Unwrap BUFFER_UNDERFLOW");
								break;
							case OK:
							}
						} catch (SSLException e) {
							isUnwrapException=true;
							e.printStackTrace();
						}
						inNetByteBuffer.compact();
						// }
						break;
					case NEED_TASK:
						Runnable task;
						while ((task = iSSLEngine.getDelegatedTask()) != null) {
							new Thread(task).start();
						}
						break;
					case FINISHED:
						break;
					case NOT_HANDSHAKING:
						//System.out.println("Not handshaking");
						break;
					}
				}
				//SelectionKey skey = keySocketChannel.keyFor(clientSelector);
				// ...
			}
			//
			if(isCloseFlag||isReadNegative1||isWriteNegative1){
				if(isCloseFlag){
					System.out.println("****** Closing...,---instanceID:"+instanceID+","+new Date());
					iSSLEngine.closeOutbound();
					while (!iSSLEngine.isOutboundDone()) {
						if (outNetByteBuffer.position() == outNetByteBuffer
								.limit()) {
							outNetByteBuffer.clear();
						}
						try {
							//SSLEngineResult resWrap = 
									iSSLEngine.wrap(
									handshakeByteBuffer, outNetByteBuffer);
						} catch (SSLException e) {
							isWrapException=true;
							e.printStackTrace();
						}

						outNetByteBuffer.flip();
						while (outNetByteBuffer.hasRemaining()) {
							int writeOutNetDataSize = 0;
							try {
								writeOutNetDataSize = socketChannel
										.write(outNetByteBuffer);
								System.out.println("<<<=== Wrote "+writeOutNetDataSize+ " bytes." );
							} catch (IOException e) {
								isWrapException=true;
								e.printStackTrace();
								break;
							}
							if (-1==writeOutNetDataSize) {
								// handle closed channel
								isWriteNegative1 = true;
								System.out.println("XXXXXX Write error.");
								break;
							}
						}
						outNetByteBuffer.compact();
					}
					try {
						socketChannel.close();
						System.out.println("****** Closed...,---instanceID:"+instanceID+","+new Date());
						break;
					} catch (IOException e) {
						
						e.printStackTrace();
					}
				}
				if(isReadNegative1||isWriteNegative1) {
					System.out.println("****** Stopping...,---instanceID:"+instanceID+","+new Date());
					
					if(0==inNetByteBuffer.position() && 0==inAppByteBuffer.position()){
						System.out.println("****** Stopped...,---instanceID:"+instanceID+","+new Date());
						break;
					}
	
				}
			}
			// linux kernel 2.2 default 100HZ
			try {
				Thread.sleep(10);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}

	}

}
